package com.quectel.taskserver.job.device;

import cn.hutool.core.lang.func.LambdaUtil;
import com.quectel.business.common.alarm.DataProcessingComponent;
import com.quectel.constant.core.alarm.AlarmConstants;
import com.quectel.constant.global.SystemConstants;
import com.quectel.core.module.device.dto.DeviceDto;
import com.quectel.core.module.device.dto.DeviceTypeDto;
import com.quectel.core.module.device.service.DeviceService;
import com.quectel.core.module.device.service.DeviceTypeService;
import com.quectel.util.common.DateUtils;
import com.quectel.util.common.JacksonUtils;
import com.quectel.util.thread.ThreadPoolUtils;
import org.apache.dubbo.config.annotation.DubboReference;
import org.apache.shardingsphere.elasticjob.api.ShardingContext;
import org.apache.shardingsphere.elasticjob.simple.job.SimpleJob;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * @author rananxin
 * @email ryan.ran@quectel.com
 * @date 2022/1/10 5:17 下午
 */
@Component
public class OfflineDetectTask implements SimpleJob, DisposableBean {

    @DubboReference
    private DeviceService deviceService;
    @DubboReference
    private DeviceTypeService deviceTypeService;
    @Autowired
    private DataProcessingComponent dataProcessingComponent;

    private static final ThreadPoolExecutor executorService = ThreadPoolUtils.createCommonIOBoundPool("offline_detect_task_pool");


    @Override
    public void destroy() throws Exception {
        ThreadPoolUtils.shutdown(executorService);
    }

    /**
     * 离线刷新 建议每三分钟执行一次
     *
     * @param shardingContext
     */
    @Override
    public void execute(ShardingContext shardingContext) {

        List<DeviceTypeDto> deviceTypeDtos = deviceTypeService.queryList(new HashMap<>());

        CompletableFuture[] futures = deviceTypeDtos.stream().map(deviceTypeDto -> CompletableFuture
                .supplyAsync(() -> offlineDetect(deviceTypeDto), executorService)
        ).toArray(CompletableFuture[]::new);

        CompletableFuture<Void> all = CompletableFuture.allOf(futures);
        try {
            all.get();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }

    }

    private boolean offlineDetect(DeviceTypeDto deviceTypeDto) {
        Integer allowLostTime = deviceTypeDto.getAllowLostTime();
        if (allowLostTime == null || allowLostTime <= 0) {
            return false;
        }
        String format = DateUtils.format(new Date(System.currentTimeMillis() - 60L * 1000 * allowLostTime), DateUtils.FormatType.COMMON);

        HashMap<String, Object> queryOnlineDevices = new HashMap<>();
        queryOnlineDevices.put(LambdaUtil.getFieldName(DeviceDto::getDeviceTypeId), String.valueOf(deviceTypeDto.getId()));
        queryOnlineDevices.put(LambdaUtil.getFieldName(DeviceDto::getOnlineStatus), SystemConstants.NORMAL.toString());
        queryOnlineDevices.put("limitTime", format);
        List<DeviceDto> offlineDevices = deviceService.queryList(queryOnlineDevices);
        for (DeviceDto deviceDto : offlineDevices) {
            HashMap<String, Object> hashMap = new HashMap<String, Object>();
            hashMap.put(AlarmConstants.DEVICE_OFFLINE_FLAG, SystemConstants.ABNORMAL);
            dataProcessingComponent.handleData(deviceDto, JacksonUtils.toJsonString(hashMap), hashMap);
        }
        return true;
    }


}
